package org.ewancle.springboothivehdfs.config;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.security.UserGroupInformation;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;

import java.io.IOException;
import java.security.PrivilegedExceptionAction;

@Slf4j
@Data
@org.springframework.context.annotation.Configuration
@ConfigurationProperties(prefix = "hadoop")
public class HadoopConfig {
    private Fs fs;
    private Auth auth;
    private Kerberos kerberos;

    @Data
    public static class Fs {
        private String defaultFS;
    }

    @Data
    public static class Auth {
        private String mode;
        private String username;
        private String password;
    }

    @Data
    public static class Kerberos {
        private String principal;
        private String keytabPath;
    }

    @Bean
    public Configuration hadoopConfiguration() {
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", fs.getDefaultFS());

        // 如果使用Kerberos认证
        if ("kerberos".equalsIgnoreCase(auth.getMode())) {
            conf.set("hadoop.security.authentication", "kerberos");
            conf.set("hadoop.security.authorization", "true");
        }

        return conf;
    }

    @Bean
    public FileSystem fileSystem(Configuration configuration) throws IOException, InterruptedException {
        if ("kerberos".equalsIgnoreCase(auth.getMode())) {
            // Kerberos认证
            log.info("Using Kerberos authentication for HDFS");
            System.setProperty("java.security.krb5.conf", "/etc/krb5.conf");
            UserGroupInformation.setConfiguration(configuration);

            UserGroupInformation ugi = UserGroupInformation.loginUserFromKeytabAndReturnUGI(
                    kerberos.getPrincipal(), kerberos.getKeytabPath());

            return ugi.doAs((PrivilegedExceptionAction<FileSystem>) () ->
                    FileSystem.get(configuration));
        } else {
            // 用户名密码认证
            log.info("Using simple authentication for HDFS with user: {}", auth.getUsername());
            UserGroupInformation ugi = UserGroupInformation.createRemoteUser(auth.getUsername());

            return ugi.doAs((PrivilegedExceptionAction<FileSystem>) () ->
                    FileSystem.get(configuration));
        }
    }
}

